/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package com.allyes.flume.source;

import static com.allyes.flume.source.TaildirSourceConfigurationConstants.*;

import java.io.File;
import java.io.FileFilter;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.regex.Pattern;

import org.apache.flume.Event;
import org.apache.flume.FlumeException;
import org.apache.flume.annotations.InterfaceAudience;
import org.apache.flume.annotations.InterfaceStability;
import org.apache.flume.client.avro.ReliableEventReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.Lists;
import com.google.common.collect.Table;
import com.google.common.collect.Table.Cell;
import com.google.gson.stream.JsonReader;

@InterfaceAudience.Private
@InterfaceStability.Evolving
public class ReliableTaildirEventReader implements ReliableEventReader {
  private static final Logger logger = LoggerFactory.getLogger(ReliableTaildirEventReader.class);

  private final Table<String, File, Pattern> tailFileTable;
  private final Table<String, String, String> staticHeaderTable;
  private final HashSet<String> dynamicHeaders;

  private TailFile currentFile = null;
  private Map<Long, TailFile> tailFiles = Collections.synchronizedMap(new HashMap<Long, TailFile>());
  private long updateTime;
  private boolean committed = true;


  /**
   * Create a ReliableTaildirEventReader to watch the given directory.
   */
  private ReliableTaildirEventReader(
      Map<String, String> filePaths,
      Table<String, String, String> headerTable,
      String positionFilePath,
      HashSet<String> dynamicHeaders) throws IOException {

    // Sanity checks
    Preconditions.checkNotNull(filePaths);
    Preconditions.checkNotNull(positionFilePath);
    Preconditions.checkNotNull(dynamicHeaders);

    if (logger.isDebugEnabled()) {
      logger.debug("Initializing {} with directory={}, metaDir={}",
          new Object[] {
              ReliableTaildirEventReader.class.getSimpleName(),
              filePaths,
              positionFilePath });
    }

    Table<String, File, Pattern> tailFileTable = HashBasedTable.create();
    for (Entry<String, String> e : filePaths.entrySet()) {
      File f = new File(e.getValue());
      File parentDir =  f.getParentFile();
      Preconditions.checkState(parentDir.exists(),
          "Directory does not exist: " + parentDir.getAbsolutePath());
      Pattern fileNamePattern = Pattern.compile(f.getName());
      tailFileTable.put(e.getKey(), parentDir, fileNamePattern);
    }
    logger.info("tailFileTable: " + tailFileTable.toString());
    logger.info("headerTable: " + headerTable.toString());

    this.tailFileTable = tailFileTable;
    this.staticHeaderTable = headerTable;
    this.dynamicHeaders = dynamicHeaders;

    updateTailFiles();

    logger.info("Updating position from position file: " + positionFilePath);
    loadPositionFile(positionFilePath);
  }

  /**
   * Load a position file which has the last read position of each file.
   * If the position file exists, update tailFiles mapping.
   */
  public void loadPositionFile(String filePath) {
    Long inode, pos;
    String path;
    FileReader fr = null;
    JsonReader jr = null;

    try {
      fr = new FileReader(filePath);
      jr = new JsonReader(fr);
      jr.beginArray();

      while (jr.hasNext()) {
        inode = null;
        pos = null;
        path = null;
        int lineIndex = 0;  // line index starts from 1

        jr.beginObject();
        while (jr.hasNext()) {
          switch (jr.nextName()) {
            case TXT_INODE:
              inode = jr.nextLong();
              break;
            case TXT_POS:
              pos = jr.nextLong();
              break;
            case TXT_LINE_INDEX:
              lineIndex = jr.nextInt();
              break;
            case TXT_FILE:
              path = jr.nextString();
              break;
          }
        }
        jr.endObject();

        for (Object v : Arrays.asList(inode, pos, path)) {
          Preconditions.checkNotNull(v, "Detected missing value in position file. "
              + "inode: " + inode + ", pos: " + pos + ", path: " + path);
        }

        TailFile tf = tailFiles.get(inode);
        if (tf != null && tf.updatePos(path, inode, pos, lineIndex)) {
          tailFiles.put(inode, tf);
        } else {
          logger.info("Missing file: " + path + ", inode: " + inode + ", pos: " + pos);
        }
      }

      jr.endArray();

    } catch (FileNotFoundException e) {
      logger.info("File not found: " + filePath + ", not updating position");

    } catch (IOException e) {
      logger.error("Failed loading positionFile: " + filePath, e);

    } finally {
      try {
        if (fr != null) {
          fr.close();
        }
        if (jr != null) {
          jr.close();
        }
      } catch (IOException e) {
        logger.error("Error: " + e.getMessage(), e);
      }
    }
  }

  public Map<Long, TailFile> getTailFiles(boolean copy) {
    if (copy) {
      Map<Long, TailFile> cpy;
      synchronized (tailFiles) {
        cpy = new HashMap<Long, TailFile>(tailFiles);
      }
      return cpy;
    } else {
      return tailFiles;
    }
  }

  public void setCurrentFile(TailFile currentFile) {
    this.currentFile = currentFile;
  }

  @Override
  public Event readEvent() throws IOException {
    List<Event> events = readEvents(1);
    if (events.isEmpty()) {
      return null;
    }
    return events.get(0);
  }

  @Override
  public List<Event> readEvents(int numEvents) throws IOException {
    return readEvents(numEvents, false);
  }

  @VisibleForTesting
  public List<Event> readEvents(TailFile tf, int numEvents) throws IOException {
    setCurrentFile(tf);
    return readEvents(numEvents, true);
  }

  public List<Event> readEvents(int numEvents, boolean backoffWithoutNL) throws IOException {
    if (!committed) {
      if (currentFile == null) {
        throw new IllegalStateException("current file does not exist. " + currentFile.getPath());
      }
      logger.info("Last read was never committed - resetting position");
      currentFile.rollBack();
    }

    List<Event> events = currentFile.readEvents(numEvents, backoffWithoutNL);
    committed = false;
    return events;
  }

  @Override
  public void close() throws IOException {
    synchronized (this.tailFiles) {
      for (TailFile tf : tailFiles.values()) {
        if (tf.getRaf() != null) {
          tf.getRaf().close();
        }
      }
    }
  }

  /** Commit the last lines which were read. */
  @Override
  public void commit() throws IOException {
    if (!committed && currentFile != null) {
      currentFile.commit(updateTime);
      committed = true;
    }
  }

  /**
   * Update tailFiles mapping if a new file is created or appends are detected
   * to the existing file.
   */
  public List<Long> updateTailFiles() throws IOException {
    updateTime = System.currentTimeMillis();
    List<Long> updatedInodes = Lists.newArrayList();

    for (Cell<String, File, Pattern> cell : tailFileTable.cellSet()) {
      Map<String, String> staticHeaders = staticHeaderTable.row(cell.getRowKey());
      File parentDir = cell.getColumnKey();
      Pattern fileNamePattern = cell.getValue();

      for (File f : getMatchFiles(parentDir, fileNamePattern)) {
        long inode = getInode(f);
        TailFile tf = tailFiles.get(inode);
        if (tf == null || !tf.getPath().equals(f.getAbsolutePath())) {
          tf = openFile(f, inode, 0, 0, staticHeaders, dynamicHeaders);
        } else {
          // the file with the same inode and name existed in 'tailFiles'(in memory)
          final boolean changed = tf.getLastUpdated() < f.lastModified();
          if (changed) {
            Pair<Long, Integer> posAndLine = tf.getCommitedPosAndLineIndex();
            if (tf.getRaf() == null) {
              tf = openFile(f, inode, posAndLine.left, posAndLine.right,
                  staticHeaders, dynamicHeaders);
            }
            if (f.length() < posAndLine.left) {
              logger.info("Pos " + posAndLine.left + " is larger than file size! " +
                  "Restarting from pos 0, file: " + tf.getPath() + ", inode: " + inode);
              tf.updatePos(tf.getPath(), inode, 0, 0);
            }
          }
          tf.setNeedTail(changed);
        }

        tailFiles.put(inode, tf);
        updatedInodes.add(inode);
      }

    }

    return updatedInodes;
  }

  private List<File> getMatchFiles(File parentDir, final Pattern fileNamePattern) {
    FileFilter filter = new FileFilter() {
      @Override
      public boolean accept(File f) {
        String fileName = f.getName();
        if (f.isDirectory() || !fileNamePattern.matcher(fileName).matches()) {
          return false;
        }
        return true;
      }
    };

    File[] files = parentDir.listFiles(filter);
    ArrayList<File> result = Lists.newArrayList(files);
    Collections.sort(result, new TailFile.CompareByLastModifiedTime());
    return result;
  }

  private long getInode(File file) throws IOException {
    long inode = (long) Files.getAttribute(file.toPath(), "unix:ino");
    return inode;
  }

  static private TailFile openFile(
      File file,
      long inode,
      long pos,
      int lineIndex,
      Map<String, String> staticHeaders,
      HashSet<String> dynamicHeaders) {

    try {
      logger.info("Opening file: " + file + ", inode: " + inode + ", pos: " +
          pos + ", line index: " + lineIndex);
      return new TailFile(file, inode, pos, lineIndex, staticHeaders, dynamicHeaders);
    } catch (IOException e) {
      throw new FlumeException("Failed opening file: " + file, e);
    }
  }

  /**
   * Special builder class for ReliableTaildirEventReader
   */
  public static class Builder {
    private Map<String, String> filePaths;
    private Table<String, String, String> headerTable;
    private String positionFilePath;
    private HashSet<String> dynamicHeaders;

    public Builder(
        Map<String, String> filePaths,
        Table<String, String, String> headerTable,
        String positionFilePath,
        HashSet<String> dynamicHeaders) {
      this.filePaths = filePaths;
      this.headerTable = headerTable;
      this.positionFilePath = positionFilePath;
      this.dynamicHeaders = dynamicHeaders;
    }

    public ReliableTaildirEventReader build() throws IOException {
      return new ReliableTaildirEventReader(
          filePaths,
          headerTable,
          positionFilePath,
          dynamicHeaders);
    }
  }

}
